package com.galeno.day02;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.Random;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/1618:00
 */
public class ParallSourceFun extends RichParallelSourceFunction<String> {
    private boolean flag=true;
    public ParallSourceFun() {
        System.out.println("构造方法执行了");
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        int index = getRuntimeContext().getIndexOfThisSubtask();
        System.out.println(index+"----Open方法被调用了");
    }

    @Override
    public void close() throws Exception {
        int index = getRuntimeContext().getIndexOfThisSubtask();
        System.out.println(index+"----close方法被调用了");
    }

    @Override
    public void run(SourceContext<String> sourceContext) throws Exception {
        int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
        System.out.println(indexOfThisSubtask +" : Run方法被调用了￥￥￥￥￥￥￥￥");
        Random random = new Random();
        //获取当前SubTask的Index

        while (flag) {
            int i = random.nextInt(100);
            sourceContext.collect(indexOfThisSubtask + " --> " + i);
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        int index = getRuntimeContext().getIndexOfThisSubtask();
        System.out.println("cancel方法被调用了");
        flag = false;

    }
}
